package main

import (
	"fmt"
	"github.com/Shopify/sarama"
	"log"
	"os"
	"time"
)

var Address = []string{"10.130.138.164:9092","10.130.138.164:9093","10.130.138.164:9094"}

func main() {
	syncProduce(Address)
}
//同步消息
func syncProduce(address []string)  {
	config := sarama.NewConfig()
	config.Producer.Return.Successes=true
	config.Producer.Timeout =5*time.Second
	p,err := sarama.NewSyncProducer(address,config)
	if err!=nil {
		log.Fatal(err)
		return
	}
	defer p.Close()
	topic:= "test"
	srcValue := "sync: this is a message. index=%d"
	for i:=0;i<10 ;i++  {
		value:=fmt.Sprintf(srcValue,i)
		msg:= &sarama.ProducerMessage{
			Topic:     topic,
			Value:     sarama.ByteEncoder(value),
		}
		part, offset, err := p.SendMessage(msg)
		if err !=nil {
			log.Fatal(err)
		}else{
			fmt.Fprintf(os.Stdout, value + "发送成功，partition=%d, offset=%d \n", part, offset)
		}
		time.Sleep(2*time.Second)
	}

}
//异步消息
func SaramaProduce()  {
	config := sarama.NewConfig()
	//等待服务器所有副本都保存成功后的响应
	config.Producer.RequiredAcks=sarama.WaitForAll
	//随机向partition发送消息(随机的分区类型)
	config.Producer.Partitioner =sarama.NewRandomPartitioner
	//是否等待成功和失败后的响应,只有上面的RequireAcks设置不是NoReponse这里才有用
	config.Producer.Return.Successes =true
	config.Producer.Return.Errors =true
	//设置使用的kafka版本,如果低于V0_10_0_0版本,消息中的timestrap没有作用.需要消费和生产同时配置
	//注意，版本设置不对的话，kafka会返回很奇怪的错误，并且无法成功发送消息
	config.Version =sarama.V2_0_0_0
	//开启
	producer,err := sarama.NewAsyncProducer([]string{"182.61.9.153:6667"},config)
	if err!=nil {
		log.Fatal(err)
		return
	}
	defer producer.Close()
	//循环判断哪个通道发送过来数据.
	fmt.Println("start goroutine")
	go func(p sarama.AsyncProducer) {
		for  {
			select {
			case <-p.Successes():
			case fail := <- p.Errors():
				fmt.Println("err: ", fail.Err)
			}
		}
		
	}(producer)

	var value1 string
	for i:=0;;i++  {
		time.Sleep(500*time.Millisecond)
		time11:=time.Now()
		//设置发送的真正内容
		value1 = "this is a message 0606 "+time11.Format("15:04:05")
		// 发送的消息,主题。
		// 注意：这里的msg必须得是新构建的变量，不然你会发现发送过去的消息内容都是一样的，因为批次发送消息的关系
		msg := &sarama.ProducerMessage{
			Topic: "0606_test",
		}
		msg.Value= sarama.ByteEncoder(value1)

		producer.Input() <- msg
	}
}